package Flink.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class BatchWordCount {
    // Todo Flink 批处理

    /**
     * 需要注意的是，这种代码的实现方式，是基于DataSet API的，也就是我们对数据的处理
     * 转换，是看作数据集来进行操作的。事实上Flink本身是流批统一的处理架构，批量的数据集
     * 本质上也是流，没有必要用两套不同的API来实现。所以从Flink 1.12开始，官方推荐的做法
     * 是直接使用DataStream API，在提交任务时通过将执行模式设为BATCH来进行批处理：
     * $ bin/flink run -Dexecution.runtime-mode=BATCH BatchWordCount.jar
     * 这样，DataSet API就没什么用了，在实际应用中我们只要维护一套DataStream API就可
     * 以。这里只是为了方便理解，依然用DataSet API做了批处理的实现。
     */
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. 从文件读取数据  按行读取(存储的元素就是每行的文本)
        DataSource<String> lineDS = env.readTextFile("Flink/input/word.txt");

        // 3. 转换数据格式
        FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Long>> collector) {
                String[] words = s.split(" ");
                for (String word : words) {
                    collector.collect(Tuple2.of(word, 1L));
                }
            }
        });

        // 4. 按照 word 进行分组
        UnsortedGrouping<Tuple2<String, Long>> wordAndOneGroupby = wordAndOne.groupBy(0);

        // 5. 分组内聚合统计
        AggregateOperator<Tuple2<String, Long>> result = wordAndOneGroupby.sum(1);

        // 6. 打印结果
        result.print();
    }
}
